package com.ruby.bigtable.hbase;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseOperation {
	
	public static HTable getTableByName(String name) throws IOException{
		Configuration configuration = HBaseConfiguration.create();
		HTable table = new HTable(configuration, name);
		return table;
	}
	
	public static void putRow(String tableName, String rowid, String username, String age, String sex) throws IOException {
		HTable table = getTableByName(tableName);
		Put put = new Put(Bytes.toBytes(rowid));
		put.add(Bytes.toBytes("info"), Bytes.toBytes("username"), Bytes.toBytes(username));
		put.add(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(age));
		put.add(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes(sex));
		table.put(put);
		table.close();
	}
	
	public static Result getResult(String tableName, String rowid) throws IOException{
		HTable table = getTableByName(tableName);
		Get get = new Get(Bytes.toBytes(rowid));
		Result result = table.get(get);
		table.close();
		return result;
	}
	
	public static void scanTable(String tableName) throws IOException{
		HTable table = getTableByName(tableName);
		Scan scan = new Scan(Bytes.toBytes("1001"));
		scan.setCacheBlocks(false);
		scan.setCaching(1000);
		scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("username"));
		ResultScanner results = table.getScanner(scan);
		for(Result result: results){
			printResult(result);
		}
	}
	
	public static void printResult(Result result){
		for(Cell cell: result.rawCells()){
			System.out.println(Bytes.toString(CellUtil.cloneRow(cell))+" "+
					Bytes.toString(CellUtil.cloneFamily(cell))+":"+
					Bytes.toString(CellUtil.cloneQualifier(cell))+"=>"+
					Bytes.toString(CellUtil.cloneValue(cell)));
		}
	}

	public static void main(String[] args) throws IOException {
		String tableName = "test:user";
		putRow(tableName, "1005", "luby", "35", "femail");
		scanTable(tableName);
	}

}
